package com.atguigu.gmall.realtime.app.func;


import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;



public class DimSink extends RichSinkFunction<JSONObject> {
    private Connection conn;
    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        //获取连接
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    //{"database":"gmall0906","data":{"tm_name":"qq","id":16},"sink_table":"dim_base_trademark","type":"insert","table":"base_trademark"}
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        //获取维度数据
        JSONObject dataJsonObj = jsonObj.getJSONObject("data");
        //获取维度表名
        String sinkTable = jsonObj.getString("sink_table");

        //拼接upsert语句
        String upsertSql = genUpsertSql(sinkTable, dataJsonObj);
        System.out.println("向Phoenix表中插入数据的SQL:" + upsertSql);

        //获取数据库操作对象
        PreparedStatement ps = null;
        try {
            ps= conn.prepareStatement(upsertSql);
            //执行SQL语句
            ps.executeUpdate();
            //注意：向Phoenix表中插入数据，需要手动提交事务。
            conn.commit();
        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("向Phoenix表中插入数据失败~~~");
        } finally {
            //释放资源
            if(ps != null){
                ps.close();
            }
        }
    }

    //dataJsonObj:{"tm_name":"qq","id":16}
    private String genUpsertSql(String tableName, JSONObject dataJsonObj) {
        Set<String> keys = dataJsonObj.keySet();
        Collection<Object> values = dataJsonObj.values();

        String upsertSql = "upsert into " + GmallConfig.PHOENIX_SCHEMA + "." + tableName
                + " (" + StringUtils.join(keys, ",") + ") values('" + StringUtils.join(values, "','") + "')";
        return upsertSql;
    }
}
